package com.simon.bigdata.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.net.URI;

/**
 * Created by changf on 2017/7/18.
 */
public class HdfsTest {

    FileSystem fs = null;

    @Before
    public void init() throws Exception {

        fs = FileSystem.get(new URI("hdfs://hdp02:9000"), new Configuration(), "root");

    }


    //从本地拷贝到hdfs
    @Test
    public void testAddFileToHdfs() throws Exception {

        //此处会删除本地文件 true
        fs.copyFromLocalFile(true, new Path("D:/1.txt"), new Path("/"));

    }


    //从hdfs拷贝到本地
    @Test
    public void testHdfsToFile() throws Exception {
        //未使用本地的文件系统 如果本地没有hdfs的相关jar java.lang.NullPointerException
        // fs.copyToLocalFile(false,new Path("/1.txt"),new Path("D:/"));
        //使用本地的文件系统
        fs.copyToLocalFile(false, new Path("/1.txt"), new Path("D:/"), true);


    }


    @Test
    public void makeDir() throws Exception {
        //fs.mkdirs(new Path("/a/b/c"));

        //这个文件就删除不掉  `/a/b is non empty': Directory is not empty
        //fs.delete(new Path("/a/b"),false);
        //现在可以删除掉
        fs.delete(new Path("/a1/b"),true);

        //fs.rename(new Path("/a"), new Path("/a1"));

    }

    //查看文件信息
    @Test
    public void listFiles() throws Exception {
        //
        //此处boolean是否迭代
        RemoteIterator<LocatedFileStatus> locatedFileStatusRemoteIterator = fs.listFiles(new Path("/"), true);
        while (locatedFileStatusRemoteIterator.hasNext()) {
            LocatedFileStatus next = locatedFileStatusRemoteIterator.next();
            System.out.println(next.getPath().getName());
            System.out.println(next.getBlockSize());
            System.out.println(next.getPermission());
            System.out.println(next.getLen());
            //获取这些文件存放的位置
            BlockLocation[] blockLocations = next.getBlockLocations();
            for (BlockLocation blockLocation:blockLocations){
                System.out.println("length:"+blockLocation.getLength());
                System.out.println("offet:"+blockLocation.getOffset());
                String[] hosts = blockLocation.getHosts();

                for (String host:hosts){
                    System.out.println("hosts:"+host);

                }
                System.out.println("=================");
            }

        }


    }


    //查看文件及文件夹信息
    @Test
    public void listFiles2() throws Exception {
        FileStatus[] fileStatuses = fs.listStatus(new Path("/"));
        for (FileStatus fileStatus:fileStatuses){
            System.out.println("fileFlag"+fileStatus.isFile());
            System.out.println("name"+fileStatus.getPath().getName());
            System.out.println("perssion"+fileStatus.getPath().getName());
        }
    }



    @Test
    public void streamAccess() throws Exception {
        FSDataOutputStream outputStream = fs.create(new Path("/angelbaby.love2"), true);
        FileInputStream inputStream=new FileInputStream("D:/3.txt");
        IOUtils.copyBytes(inputStream,outputStream,new Configuration());
    }


    /**
     * 字节读取
     * @throws Exception
     */

    @Test
    public void testDownLoadToLocal() throws Exception {
        FSDataInputStream inputStream = fs.open(new Path("/angelbaby.love"));
        FileOutputStream outputStream=new FileOutputStream("D:/2.txt");
        IOUtils.copyBytes(inputStream,outputStream,4096);
    }



    //自由获取
    @Test
    public void testRandomAccess() throws Exception {
        //1234567890abcdefghijklmn
        FSDataInputStream inputStream = fs.open(new Path("/angelbaby.love2"));
        inputStream.seek(2);
        FileOutputStream outputStream=new FileOutputStream("D:/5.txt");
        //【以下俩种的运行结果不一致】
        //34567
        IOUtils.copyBytes(inputStream,outputStream,5L,true);
        //34567890abcdefghijklmn
        IOUtils.copyBytes(inputStream,outputStream,5,true);
    }


    //获取文件的block信息，读取指定block内容
    @Test
    public void testCat()throws Exception{
        FSDataInputStream inputStream = fs.open(new Path("/angelbaby.love2"));
        FileStatus[] fileStatuses = fs.listStatus(new Path("/angelbaby.love2"));
        BlockLocation[] fileBlockLocations = fs.getFileBlockLocations(new Path("/angelbaby.love2"), 0, fileStatuses[0].getLen());
        //获取block的长度
        long length = fileBlockLocations[0].getLength();
        //获取偏移量
        long offset = fileBlockLocations[0].getOffset();
        System.out.println("length:"+length+";offset:"+offset);

        IOUtils.copyBytes(inputStream,System.out,(int)length);

        byte[] bytes=new byte[10];

        FileOutputStream os=new FileOutputStream(new File("D:/7.txt"));
        while(inputStream.read(offset,bytes,0,10)!=-1){
            os.write(bytes);
            System.out.println("偏移量："+offset);
            System.out.println("数据："+new String(bytes));
            offset+=10;
            if(offset>length)return;
        }
        os.flush();
        os.close();
        inputStream.close();

    }

    @After
    public void close() throws Exception {
        fs.close();

    }
}
